package com.amos.think.service.mq;

import com.amos.think.domain.gateway.UserGateway;
import com.amos.think.domain.user.UserEntity;
import com.amos.think.dto.UserRegisterCmd;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class RocketMQConsumer {
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Value("${rocketmq.consumer.topic}")
    private String topic;

    private DefaultMQPushConsumer consumer;

    @Autowired
    private UserGateway userGateway;

    private ObjectMapper objectMapper;

    public RocketMQConsumer(){
        objectMapper = new ObjectMapper();
    }


    public void start() throws Exception {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe(topic, "*");
        //消费者监听,生产者发送消息,及时处理
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                // 处理消息
                String messageBody  = new String(msg.getBody());
                try {
                    log.info("Receive message:{}", new String(msg.getBody()));
                    UserEntity userEntity = objectMapper.readValue(messageBody, UserEntity.class);
                    userGateway.save(userEntity);
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }

    public void shutdown() {
        if (consumer != null) {
            consumer.shutdown();
        }
    }

}
